//package com.example.demo.service;
//
//import org.springframework.beans.factory.DisposableBean;
//import org.springframework.data.redis.connection.stream.*;
//import org.springframework.data.redis.core.StringRedisTemplate;
//import org.springframework.data.redis.core.StreamOperations;
//import org.springframework.stereotype.Service;
//
//import java.time.Duration;
//import java.util.List;
//import java.util.concurrent.*;
//import java.util.concurrent.atomic.AtomicLong;
//
///**
// * @author vaintale
// * @date 2025/7/3
// */
//@Service
//public class RedisStreamConsumerServiceBak implements DisposableBean {
//    private final String streamKey = "redis-stream-01";
//    private final String group = "group-alg";
//    private final String consumer = "consumer-1";
//    private final StringRedisTemplate redisTemplate;
//    private final StreamOperations<String, Object, Object> streamOps;
//    private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
//    private final AtomicLong intervalMillis = new AtomicLong(1000);
//    private ScheduledFuture<?> future;
//
//    public RedisStreamConsumerServiceBak(StringRedisTemplate redisTemplate) {
//        this.redisTemplate = redisTemplate;
//        this.streamOps = redisTemplate.opsForStream();
//        createGroupIfNotExists();
//        start();
//    }
//
//    private void createGroupIfNotExists() {
//        try {
//            // 创建消费组（如果不存在）
//            streamOps.createGroup(streamKey, ReadOffset.latest(), group);
//        } catch (Exception e) {
//            // group已存在时会报错，忽略
//        }
//    }
//
//    public void setIntervalMillis(long millis) {
//        intervalMillis.set(millis);
//        start(); // 重新调整调度频率
//    }
//
//    public void start() {
//        if (future != null && !future.isCancelled()) {
//            future.cancel(true);
//        }
//        future = executor.scheduleWithFixedDelay(this::consume, 0, intervalMillis.get(), TimeUnit.MILLISECONDS);
//    }
//
//    private void consume() {
//        try {
//            // 读取消息（一次最多1条，可调整）
//            List<MapRecord<String, Object, Object>> messages = streamOps.read(
//                    Consumer.from(group, consumer),
//                    StreamReadOptions.empty().count(1).block(Duration.ofMillis(100)),
//                    StreamOffset.create(streamKey, ReadOffset.lastConsumed())
//            );
//            if (messages != null) {
//                for (MapRecord<String, Object, Object> message : messages) {
//                    // 处理消息
//                    System.out.println("收到消息: " + message.getId() + " " + message.getValue());
//                    // 手动ACK
//                    streamOps.acknowledge(streamKey, group, message.getId());
//                    streamOps.delete(streamKey, message.getId());
//                }
//            }
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//
//    @Override
//    public void destroy() {
//        executor.shutdownNow();
//    }
//}
